using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Me.Prettyprint.Cassandra.Connection;
using Me.Prettyprint.Cassandra.Model;
using Me.Prettyprint.Cassandra.Model.Thrift;
using Me.Prettyprint.Cassandra.Serializers;
using Me.Prettyprint.Cassandra.Service;
using Me.Prettyprint.Cassandra.Utils;
using Me.Prettyprint.Hector.Api;
using Me.Prettyprint.Hector.Api.Beans;
using Me.Prettyprint.Hector.Api.Factory;
using Me.Prettyprint.Hector.Api.Query;
using Org.Apache.Cassandra.Thrift;

namespace Me.Prettyprint.Cassandra.Service
{
    /// <summary>
    /// Iterates over the MultigetSliceQuery result set, refreshing until all
    /// qualifying rows are retrieved based on input keys.&nbsp; This iterator is
    /// optimized for parallelism with the help of maxThreadCount option provided. If
    /// maxThreadCount is not provided, it calls Cassandra with the
    /// set(maxRowCountPerQuery) of row keys at a time unless all keys are queried.
    /// E.g., maxRowCountPerQuery is 100 and maxThreadCount 5, it calls Cassandra 5
    /// times using 5 threads/ parallelism for total of 500 keys. You can also
    /// configure it not to use Threads and call Cassandra 5 time sequentially
    /// instead of parallelism by not setting maxThreadCount or setting it 0 
    /// </summary>
    /// <typeparam name="K"></typeparam>
    /// <typeparam name="N"></typeparam>
    /// <typeparam name="V"></typeparam>
    public class MultigetSliceEnumerator<K, N, V> : IEnumerator<Row<K, N, V>>
	{
        private const int DEFAULT_MAXCOL_COUNT = 100;
        private const int DEFAULT_MAXROW_COUNT_PERQUERY = 0;
        private const int DEFAULT_MAX_THREAD_COUNT = 0;
        private IEnumerator<Row<K, N, V>> enumerator;
        private N start;
        private N finish;
        private bool reversed;
        private int rowKeysIndex = 0;
        private List<List<K>> rowKeysList = new List<List<K>>();
        private List<K> rowKeys = new List<K>();
        private int maxRowCountPerQuery = DEFAULT_MAXROW_COUNT_PERQUERY;
        private Keyspace keyspace;
        private Serializer<K> keySerializer;
        private Serializer<N> nameSerializer;
        private Serializer<V> valueSerializer;
        private String columnFamily;
        private AtomicLong  totalExecutionTimeMicro = new AtomicLong(0);
        private AtomicLong totalExecutionTimeNano = new AtomicLong(0);
        private int maxThreads = DEFAULT_MAX_THREAD_COUNT;
        private int maxColumnCount = DEFAULT_MAXCOL_COUNT;
        private int threadCount = DEFAULT_MAX_THREAD_COUNT;
        private int numKeysPerThread; 
        private Dictionary<string,CassandraHost> m_hostsUsed = new Dictionary<string,CassandraHost>();
        private List<Rows<K, N, V>> queryResult = new List<Rows<K, N, V>>();

        public MultigetSliceEnumerator(bool reversed, Keyspace keyspace,
			Serializer<K> keySerializer, Serializer<N> nameSerializer,
			Serializer<V> valueSerializer, String columnFamily,
			List<K> rowKeys, N start, N finish) :
            this(reversed, keyspace, keySerializer, nameSerializer,
                valueSerializer, columnFamily, rowKeys, start, finish,
                DEFAULT_MAX_THREAD_COUNT, DEFAULT_MAXROW_COUNT_PERQUERY,
                DEFAULT_MAXCOL_COUNT)
        {
        }
        public MultigetSliceEnumerator(bool reversed, Keyspace keyspace,
			Serializer<K> keySerializer, Serializer<N> nameSerializer,
			Serializer<V> valueSerializer, String columnFamily,
			List<K> rowKeys, N start, N finish, int maxColumnCountPerRow) :
            this(reversed, keyspace, keySerializer, nameSerializer,
            valueSerializer, columnFamily, rowKeys, start, finish,
            DEFAULT_MAX_THREAD_COUNT, DEFAULT_MAXROW_COUNT_PERQUERY, maxColumnCountPerRow)
        {
        }
        public MultigetSliceEnumerator(bool reversed, int maxRowCountPerQuery,
			Keyspace keyspace, Serializer<K> keySerializer,
			Serializer<N> nameSerializer, Serializer<V> valueSerializer,
			String columnFamily, List<K> rowKeys, N start, N finish,
			int maxColumnCountPerRow) :
        this(reversed, keyspace, keySerializer, nameSerializer,
				valueSerializer, columnFamily, rowKeys, start, finish,
				DEFAULT_MAX_THREAD_COUNT, maxRowCountPerQuery,
				maxColumnCountPerRow)
        {
        }
        public MultigetSliceEnumerator(bool reversed, int maxRowCountPerQuery,
			Keyspace keyspace, Serializer<K> keySerializer,
			Serializer<N> nameSerializer, Serializer<V> valueSerializer,
			String columnFamily, List<K> rowKeys, N start, N finish) :
        this(reversed, keyspace, keySerializer, nameSerializer,
				valueSerializer, columnFamily, rowKeys, start, finish,
				DEFAULT_MAX_THREAD_COUNT, maxRowCountPerQuery,
				DEFAULT_MAXCOL_COUNT)
        {
        }
        public MultigetSliceEnumerator(bool reversed, Keyspace keyspace,
			Serializer<K> keySerializer, Serializer<N> nameSerializer,
			Serializer<V> valueSerializer, String columnFamily,
			List<K> rowKeys, N start, N finish, int maxThreadCount,
			int maxRowCountPerQuery, int maxColumnCountPerRow)
        {
            this.reversed = reversed;
            this.maxRowCountPerQuery = maxRowCountPerQuery;
            this.keyspace = keyspace;
            this.keySerializer = keySerializer;
            this.nameSerializer = nameSerializer;
            this.valueSerializer = valueSerializer;
            this.columnFamily = columnFamily;
            this.start = start;
            this.finish = finish;
            this.rowKeys = rowKeys;
            this.maxColumnCount = maxColumnCountPerRow;
            this.maxThreads = maxThreadCount;
            this.rowKeysList = PrepareKeysForParallelism();
        }
        public bool MoveNext()
        {
            if (enumerator == null)
            {
                this.RunQuery();
            } else if (!enumerator.MoveNext() && rowKeysIndex < this.rowKeysList.Count)
            {
                this.RunQuery();
            }
            return enumerator.MoveNext();
        }
        public new Row<K, N, V> Current
        {
            get { return enumerator.Current; }
        }
        object System.Collections.IEnumerator.Current
        {
            get { return this.Current; }
        }
        public void Dispose()
        {
        }
        public void Reset()
        {
        }
        private List<List<K>> PrepareKeysForParallelism() 
        {
            List<List<K>> returnKeys = new List<List<K>>();
            int numKeys = rowKeys.Count;
            int numThreads = 1;
            
            if(maxRowCountPerQuery > 0)
            {
                numThreads=(int) Math.Ceiling((numKeys / (double) maxRowCountPerQuery));
                numThreads=	Math.Max(numThreads, 1);
            }
            threadCount=Math.Min(numThreads, maxThreads);
            
            numKeysPerThread = (int) Math.Ceiling(numKeys / (double) numThreads);
            numKeysPerThread = Math.Max(numKeysPerThread, 1);
            
            if(this.rowKeys != null && this.rowKeys.Count > 0)
            {
                returnKeys = List.Partition(rowKeys, numKeysPerThread);
            }
            return returnKeys;
        }
        private async void RunQuery()
        {
            if(this.rowKeysList != null && this.rowKeysList.Count > 0)
            {
                await Task.Factory.StartNew( () =>
                    {
                        foreach (List<K> keylist in this.rowKeysList)
                        {
							RunMultigetSliceQuery(keylist); 
                        }
                    }, CancellationToken.None, TaskCreationOptions.None, TaskScheduler.FromCurrentSynchronizationContext());
                // set the rowKeysIndex to size of input keys so as no further calls
				// will be made to Cassandra.
				// This ensures iterator.hasNext() returns false when all keys are
				// queried
				rowKeysIndex = this.rowKeysList.Count;

			}
            List<Row<K, N, V>> resultList = new List<Row<K, N, V>>(queryResult.Count);
            lock (this.queryResult)
            {
                // Ensure that runMultigetSliceQuery() method call updates global
                // variable queryResult with query result (if exists)
                if (queryResult != null && queryResult.Count > 0)
                {
                    foreach (Rows<K, N, V> rows in queryResult)
                    {
                        if (rows != null && rows.Count > 0)
                        {
                            foreach (Row<K, N, V> row in rows)
                            {
                                // prepare List<Row<K, N, V>> to return
                                // the iterator of <Row<K,N,V>> to the caller
                                resultList.Add(row);
                            }
                        }
                    }
                }
                // assign global iterator with the result of multigetSliceQuery
                //iterator = resultList.iterator();
                this.enumerator = resultList.GetEnumerator();
            }
        }
        private void RunMultigetSliceQuery(List<K> param)
        {
            MultigetSliceQuery<K, N, V> multigetSliceQuery = HFactory.CreateMultigetSliceQuery(keyspace, keySerializer,	nameSerializer, valueSerializer);
            multigetSliceQuery.SetColumnFamily(columnFamily);
            multigetSliceQuery.SetKeys(param);
            multigetSliceQuery.SetRange(start, finish, reversed, maxColumnCount);
            QueryResult<Rows<K, N, V>> result = multigetSliceQuery.Execute();
            
            queryResult.Add(result.Get());
            // Add current query execution time to internal variable
            // totalExecutionTimeMicro. When parallelism is enabled, this value might not be correct 
            // always due to available system resources and thread implementation
            totalExecutionTimeMicro.AddAndGet(result.ExecutionTimeMicro);
            // Add current query execution time to internal variable
            // totalExecutionTimeNano. When parallelism is enabled, this value might not be correct 
            // always due to available system resources and thread implementation
            totalExecutionTimeNano.AddAndGet(result.ExecutionTimeNano);
            
            //Add host used to the list
            m_hostsUsed[result.HostUsed.Ip] = result.HostUsed;
        }
	}
}